package com.alan.netty.rpc.v1;

import com.alan.netty.rpc.common.RpcRequest;
import com.alan.netty.rpc.common.RpcResponse;
import com.alan.netty.rpc.common.SerializerFactory;
import com.alan.netty.rpc.common.handler.RpcDecode;
import com.alan.netty.rpc.common.handler.RpcEncode;
import com.alan.netty.rpc.common.serialize.ProtosuffSerialize;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;

/**
 * @author weiBin
 * @date 2018/11/27
 */
@Slf4j
public class RpcServer {
    private int port;
    private Thread thread;

    public RpcServer(int port) {
        this.port = port;
    }

    public void start(RpcBuilder builder) {
        thread = new Thread(() -> {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            EventLoopGroup bossGroup = new NioEventLoopGroup(1);
            NioEventLoopGroup workerGroup = new NioEventLoopGroup(8);
            try {
                serverBootstrap.group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .childHandler(new ChannelInitializer<NioSocketChannel>() {
                            @Override
                            protected void initChannel(NioSocketChannel channel) throws Exception {
                                ChannelPipeline pipeline = channel.pipeline();
                                //解码rpcRequest
                                pipeline.addLast(new RpcDecode(RpcRequest.class, SerializerFactory.getSerializer()));
                                //编码RPCResponse
                                pipeline.addLast(new RpcEncode(RpcResponse.class, SerializerFactory.getSerializer()));
                                //对解码后的rpcRequest 进行业务处理
                                pipeline.addLast(new RpcServerHandler(builder));
                            }
                        })
                        .option(ChannelOption.SO_BACKLOG, 128)
                        .option(ChannelOption.SO_REUSEADDR, true)
                        .childOption(ChannelOption.TCP_NODELAY, true)
                        .childOption(ChannelOption.SO_KEEPALIVE, true);
                ChannelFuture future = serverBootstrap.bind(port).sync();
                log.info("服务器启动成功,开始监听" + port);
                future.channel().closeFuture().sync();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        });
        thread.start();
    }

    public Thread getThread() {
        return this.thread;
    }

    public void stop() throws Exception {
        if (thread != null && thread.isAlive()) {
            thread.interrupt();
        }

        log.info("接口服务器端停止......");
    }

}
